package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class Demo03MySource {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> mySourceDS = env.addSource(new MySource());
        mySourceDS.print();

        env.execute();


    }
}

class MySource implements SourceFunction<String>{

    // Source启动时会执行
    // run方法如果会结束，则Source会得到一个有界流
    // run方法如果不会结束，则Source会得到一个无界流
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        System.out.println("run方法启动了");
        // ctx 可以通过collect方法向下游发送数据
        long cnt = 0L;
        while(true){
            ctx.collect(cnt+"");
            cnt ++;
            // 休眠一会
            Thread.sleep(1000);
        }


    }

    // Source结束时会执行
    @Override
    public void cancel() {
        System.out.println("Source结束了");

    }
}
